package com.dt.platform.ops.service.impl;

import com.alibaba.csp.sentinel.util.StringUtil;
import com.alibaba.druid.pool.DruidDataSource;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.dt.platform.constants.enums.ops.MonitorMethodEnum;
import com.dt.platform.domain.ops.MonitorNode;
import com.dt.platform.domain.ops.MonitorTplIndicator;
import com.dt.platform.ops.service.IMonitorDataProcessJdbcService;
import com.github.foxnic.api.error.ErrorDesc;
import com.github.foxnic.api.transter.Result;
import com.github.foxnic.commons.busi.id.IDGenerator;
import com.github.foxnic.commons.concurrent.SimpleJoinForkTask;
import com.github.foxnic.commons.encrypt.AESUtil;
import com.github.foxnic.commons.encrypt.MD5Util;
import com.github.foxnic.commons.log.Logger;
import com.github.foxnic.dao.data.RcdSet;
import com.github.foxnic.dao.spec.DAO;
import com.github.foxnic.dao.spec.DAOBuilder;
import com.github.foxnic.sql.expr.Insert;
import com.github.foxnic.sql.meta.DBType;
import org.github.foxnic.web.framework.dao.DBConfigs;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import javax.annotation.Resource;
import javax.sql.DataSource;
import java.util.ArrayList;
import java.util.Date;
import java.util.List;

@Service("MonitorDataProcessJdbcServiceImpl")
public class MonitorDataProcessJdbcServiceImpl implements IMonitorDataProcessJdbcService {
    /**
     * 注入DAO对象
     * */
    @Resource(name= DBConfigs.PRIMARY_DAO)
    private DAO dao=null;

    /**
     * 获得 DAO 对象
     * */
    public DAO dao() { return dao; }




    @Autowired
    private MonitorDataProcessBaseServiceImpl monitorDataProcessBaseService;

    @Override
    public Result collectData() {

        //后期可以继续优化
        List<MonitorNode> nodeList=new ArrayList<>();
        List<MonitorNode> nodeListDb2=monitorDataProcessBaseService.queryExecuteNodeList(MonitorMethodEnum.DB2_JDBC.code());
        List<MonitorNode> nodeListOracle=monitorDataProcessBaseService.queryExecuteNodeList(MonitorMethodEnum.ORACLE_JDBC.code());
        List<MonitorNode> nodeListMysql=monitorDataProcessBaseService.queryExecuteNodeList(MonitorMethodEnum.MYSQL_JDBC.code());
        List<MonitorNode> nodeListPg=monitorDataProcessBaseService.queryExecuteNodeList(MonitorMethodEnum.PG_JDBC.code());
        List<MonitorNode> nodeListSqlServer=monitorDataProcessBaseService.queryExecuteNodeList(MonitorMethodEnum.SQLSERVER_JDBC.code());
        List<MonitorNode> nodeListSqlGaussDB=monitorDataProcessBaseService.queryExecuteNodeList(MonitorMethodEnum.GAUSSDB_JDBC.code());
        List<MonitorNode> nodeListDmDB=monitorDataProcessBaseService.queryExecuteNodeList(MonitorMethodEnum.DM_JDBC.code());
        if(nodeListDmDB!=null&&nodeListDmDB.size()>0){
            nodeList.addAll(nodeListDmDB);
        }
        if(nodeListDb2!=null&&nodeListDb2.size()>0){
            nodeList.addAll(nodeListDb2);
        }
        if(nodeListOracle!=null&&nodeListOracle.size()>0){
            nodeList.addAll(nodeListOracle);
        }
        if(nodeListSqlGaussDB!=null&&nodeListSqlGaussDB.size()>0){
            nodeList.addAll(nodeListSqlGaussDB);
        }
        if(nodeListOracle!=null&&nodeListOracle.size()>0){
            nodeList.addAll(nodeListMysql);
        }
        if(nodeListPg!=null&&nodeListPg.size()>0){
            nodeList.addAll(nodeListPg);
        }
        if(nodeListSqlServer!=null&&nodeListSqlServer.size()>0){
            nodeList.addAll(nodeListSqlServer);
        }
        Logger.info("collectData,find node number:"+nodeList.size());
        // 创建 ForkJoin 工具，其中 输入一个 Integer 元素的 List ，输出 Long 元素的 List ，每个线程处理 若干元素 ，此处为 5 个
        if(nodeList.size()>0) {
            int batchSize=monitorDataProcessBaseService.calBathSizeValue(nodeList.size());
            Logger.info("current batch size:"+batchSize);
            SimpleJoinForkTask<MonitorNode, Result> task = new SimpleJoinForkTask<>(nodeList, batchSize);
            // 并行执行
            List<Result> rvs = task.execute(els -> {
                // 打印当前线程信息
                Logger.info(Thread.currentThread().getName());
                // 处理当前分到的 若干元素，此处为 5 个
                List<Result> rs = new ArrayList<>();
                for (MonitorNode node : els) {
                    Result result = collectNodeData(node, null, false);
                    if(result.isSuccess()){
                        Logger.debug("node:"+node.getNodeNameShow()+",result:"+result.success()+",message:"+result.getMessage());
                    }else{
                        Logger.error("node:"+node.getNodeNameShow()+",result:"+result.success()+",message:"+result.getMessage());
                    }
                    rs.add(result);
                }
                // 处理完毕，返回本批次的处理结果，由 SimpleJoinForkTask 负责合并全部结果，并由 rvs 变量接收
                return rs;
            });

        }

        return ErrorDesc.success();
    }


    //同一个节点指标必须一样
    @Override
    public Result collectNodeData(MonitorNode node,String indicatorId,boolean isForce) {

        String sql="select distinct c.monitor_method from ops_monitor_tpl a,ops_monitor_node_tpl_item b,ops_monitor_tpl_indicator c\n" +
                "where a.code=b.tpl_code \n" +
                "and a.code=c.monitor_tpl_code\n" +
                "and a.deleted=0\n" +
                "and c.deleted=0\n" +
                "and a.status='enable'\n" +
                "and c.status='enable'\n" +
                "and c.monitor_method in ('oracle_jdbc','db2_jdbc','mysql_jdbc','sqlserver_jdbc','gaussdb_jdbc','pg_jdbc')\n" +
                "and b.node_id=?";
        dao.pausePrintThreadSQL();
        RcdSet rs=dao.query(sql,node.getId());
        dao().resumePrintThreadSQL();
        if(rs.size()==0){
            return ErrorDesc.success();
        }

        if(rs.size()!=1){
            return ErrorDesc.failureMessage("当前节点,存在超过2个以上jdbc类型,只支持一个");
        }

        String var=node.getVar();
        JSONObject varObj=JSONObject.parseObject(var);
        if(var==null){
            return ErrorDesc.failureMessage("当前节点,所需的变量不存在");
        }
        String user=varObj.getString("user");
        String pwd=varObj.getString("pwd");
        if(varObj.containsKey("user")&&varObj.containsKey("pwd")){
            if(varObj.containsKey("encrypt")&& "true".equals(varObj.getString("encrypt"))){
                AESUtil aes=new AESUtil("collect");
                pwd=aes.decryptData(varObj.getString("pwd"));
            }
        }else{
            Logger.info("请输入连接的账户、密码信息");
            return ErrorDesc.failureMessage("请输入连接的账户、密码信息");
        }
        String me=rs.getRcd(0).getString("monitor_method");
        List<MonitorTplIndicator> monitorTplIndicatorList=monitorDataProcessBaseService.queryExecuteIndicatorList(node.getId(),me,null,isForce);
        if(monitorTplIndicatorList==null||monitorTplIndicatorList.size()==0){
            Logger.info("collect method:jdbc,node:"+node.getNodeNameShow()+",当前没有指标触发采集");
            return ErrorDesc.success();
        }
        String dbType="";
        String driver="";
        if(MonitorMethodEnum.DB2_JDBC.code().equals(me)){
            dbType="db2";
            driver="com.ibm.db2.jcc.DB2Driver";
        }else if(MonitorMethodEnum.ORACLE_JDBC.code().equals(me)){
            dbType="oracle";
            driver="oracle.jdbc.driver.OracleDriver";
        }else if(MonitorMethodEnum.MYSQL_JDBC.code().equals(me)||MonitorMethodEnum.OB_MYSQL_JDBC.code().equals(me)){
            dbType="mysql"; //mysql 5.7
            driver="com.mysql.jdbc.Driver";
        }else if(MonitorMethodEnum.SQLSERVER_JDBC.code().equals(me)){
            dbType="sqlserver";
            driver="com.microsoft.sqlserver.jdbc.SQLServerDriver";
        }else if(MonitorMethodEnum.PG_JDBC.code().equals(me)||MonitorMethodEnum.GAUSSDB_JDBC.code().equals(me)){
            dbType="pg";
            driver="org.postgresql.Driver";
        }else if(MonitorMethodEnum.DM_JDBC.code().equals(me)){
            dbType="dm";
            driver="dm.jdbc.driver.DmDriver";
        }else{
            return ErrorDesc.failureMessage("当前节点所选JDBC不支持");
        }
        if(varObj.containsKey("driver")&& ( !StringUtil.isBlank(varObj.getString("driver")) )){
            driver=varObj.getString("driver");
        }
        Logger.info("node:"+node.getNodeNameShow()+",dbType:"+dbType+",driver:"+driver);

        DruidDataSource ds=createDataSource(dbType,driver,node.getJdbcUrl(),user,pwd);
        DAOBuilder builder=new DAOBuilder();
        DAO newDao=null;
        try {
            newDao=builder.datasource(ds).build();
            newDao.setPrintSQL(true);
            newDao.setPrintSQLSimple(true);
            newDao.setPrintSQLCallstack(false);
        } catch (Exception e) {
            e.printStackTrace();
            return ErrorDesc.failureMessage("创建datasource失败");
        }

        if(newDao==null){
            Logger.info("create dao failed!,url:"+node.getJdbcUrl()+",user:"+user+",pwd:"+( StringUtil.isBlank(pwd)?"none":MD5Util.encrypt16(pwd)));
            //插入节点报错数据
            MonitorTplIndicator indicator=monitorTplIndicatorList.get(0);
            Insert errInsert=new Insert("ops_monitor_node_value");
            errInsert.set("id", IDGenerator.getSnowflakeId());
            errInsert.setIf("indicator_code",indicator.getCode());
            errInsert.setIf("node_id",node.getId());
            if("db.connect".equals(indicator.getCode())){
                errInsert.setIf("is_connected",0);
                errInsert.setIf("result_status","sucess");
                errInsert.setIf("result_message","dao create failed");
            }else{
                errInsert.setIf("is_connected",1);
                errInsert.setIf("result_status","failed");
                errInsert.setIf("result_message","dao create failed");
            }
            errInsert.setIf("monitor_tpl_code",indicator.getMonitorTplCode());
            errInsert.setIf("record_time",new Date());
            this.dao().execute(errInsert);
            errInsert.into("ops_monitor_node_value_last");
            this.dao().execute(errInsert);
            return ErrorDesc.failureMessage("dao创建失败");
        }else{
            Logger.info("create dao success!"+",url:"+node.getJdbcUrl());
        }
        List<Insert> list=queryIndicators(newDao,monitorTplIndicatorList,node);
        if(list.size()>0){
            monitorDataProcessBaseService.executeCollectDataInsert(list,"collect by jdbc");
        }
        if (ds != null && !ds.isClosed()) {
            ds.close();
            Logger.info(node.getJdbcUrl()+",DruidDataSource已关闭，所有连接已释放。");
        }
        return ErrorDesc.success();

    }

    public List<Insert> queryIndicators(DAO d,List<MonitorTplIndicator> indicators,MonitorNode node){
        List<Insert> insertList=new ArrayList<>();
        for(int i=0;i<indicators.size();i++){
            List<Insert> ins=queryIndicatorData(d,indicators.get(i),node);
            if(ins!=null&&ins.size()>0){
                insertList.addAll(ins);
            }
        }
        return insertList;
    }
    public List<Insert> queryIndicatorData(DAO dbDao,MonitorTplIndicator indicator,MonitorNode node){
        List<Insert> list=new ArrayList<>();
        if(!indicator.getStatus().equals("enable")){
            return list;
        }
        RcdSet rs=null;
        try{
            //{varList:['db','type']},用来定义具体的变量名称
            String indVar=indicator.getIndicatorVariable();
            //从node只获取具体的变量名称
            String var=node.getVar();
            String cmd=indicator.getCommand();
            if(StringUtil.isBlank(var)||StringUtil.isEmpty(indVar)){
                Logger.info("node:"+node.getNodeNameShow()+",var is blank,indicator:"+indicator.getIndicatorVariable()+"!");
            }else{
                JSONObject varJson = JSONObject.parseObject(var);
                JSONObject indVarObj= JSONArray.parseObject(indVar);
                JSONArray indVarJson= new JSONArray();
                if(indVarObj.containsKey("varList")){
                    indVarJson=indVarObj.getJSONArray("varList");
                }
                if(indVarJson.size()>0&&varJson!=null){
                    for(int i=0;i<indVarJson.size();i++){
                        String key=indVarJson.getString(i).trim();
                        if(varJson.containsKey(key)){
                            String keyValue=varJson.getString(key);;
                            cmd=cmd.replace("${"+key+"}",keyValue);
                        }
                    }
                }
            }

            //只能执行select、show开头的命令
            boolean ckCmdOK=false;
            String chkCmdMsg="";
            if(StringUtil.isBlank(cmd)){
                ckCmdOK=false;
                chkCmdMsg="执行内容为空";
            }else{
                if(cmd.trim().toLowerCase().startsWith("select")
                        ||cmd.trim().toLowerCase().startsWith("call")
                        ||cmd.trim().toLowerCase().startsWith("show")
                ){
                    chkCmdMsg="检查通过";
                    ckCmdOK=true;
                }else{
                    chkCmdMsg="执行内容正确";
                    ckCmdOK=false;
                }
            }

            if(!ckCmdOK){
                Insert valueInsert2=new Insert("ops_monitor_node_value");
                valueInsert2.set("id", IDGenerator.getSnowflakeId());
                valueInsert2.setIf("result_status","failed");
                valueInsert2.setIf("result_message",chkCmdMsg);
                valueInsert2.setIf("indicator_code",indicator.getCode());
                valueInsert2.setIf("node_id",node.getId());
                valueInsert2.setIf("is_connected",1);
                valueInsert2.setIf("monitor_tpl_code",indicator.getMonitorTplCode());
                valueInsert2.setIf("record_time",new Date());
                list.add(valueInsert2);
                return list;
            }

            rs=dbDao.query(cmd);
            if(rs==null||rs.size()==0){
                Logger.error("node:"+node.getNodeNameShow()+",indicator:"+indicator.getCode()+",query result size null");
                Insert valueInsert=new Insert("ops_monitor_node_value");
                valueInsert.set("id", IDGenerator.getSnowflakeId());
                valueInsert.setIf("result_status","failed");
                valueInsert.setIf("result_message","result is null");
                valueInsert.setIf("indicator_code",indicator.getCode());
                valueInsert.setIf("node_id",node.getId());
                valueInsert.setIf("is_connected",1);
                valueInsert.setIf("monitor_tpl_code",indicator.getMonitorTplCode());
                valueInsert.setIf("record_time",new Date());
                list.add(valueInsert);
            }else{
                String[] mapArr=indicator.getValueColumnMap().split(",");
                String content="";
                for(int i=0;i<rs.size();i++){
                    String e="";
                    for(int j=0;j<mapArr.length;j++){
                        String key=mapArr[j];
                        String value=rs.getRcd(i).getString(key);
                        if(j==0){
                            e=rs.getRcd(i).getString(key);
                        }else{
                            e=e+","+value;
                        }
                    }
                    if(i==0){
                        content=e;
                    }else{
                        content=content+"\n"+e;
                    }
                }
                Logger.info("collect content:"+content);
                Result<Object> insertResult=monitorDataProcessBaseService.convertToInsertData(indicator,content,node);
                if(insertResult.isSuccess()){
                    list=(List<Insert>)insertResult.getData();
                }else{
                    Insert valueInsert2=new Insert("ops_monitor_node_value");
                    valueInsert2.set("id", IDGenerator.getSnowflakeId());
                    valueInsert2.setIf("result_status","failed");
                    valueInsert2.setIf("result_message",insertResult.getMessage());
                    valueInsert2.setIf("indicator_code",indicator.getCode());
                    valueInsert2.setIf("node_id",node.getId());
                    valueInsert2.setIf("is_connected",1);
                    valueInsert2.setIf("monitor_tpl_code",indicator.getMonitorTplCode());
                    valueInsert2.setIf("record_time",new Date());
                    list.add(valueInsert2);
                }
            }
        }catch (Exception e){
            e.printStackTrace();
            Insert valueInsert2=new Insert("ops_monitor_node_value");
            valueInsert2.set("id", IDGenerator.getSnowflakeId());
            valueInsert2.setIf("result_status","failed");
            valueInsert2.setIf("result_message",e.getMessage().length()>1024?e.getMessage().substring(0,1000):e.getMessage());
            valueInsert2.setIf("indicator_code",indicator.getCode());
            valueInsert2.setIf("node_id",node.getId());
            valueInsert2.setIf("is_connected",1);
            valueInsert2.setIf("monitor_tpl_code",indicator.getMonitorTplCode());
            valueInsert2.setIf("record_time",new Date());
            list.add(valueInsert2);
        }
        return list;
    }


    private DruidDataSource createDataSource(String type, String driverName, String url, String userName, String passwd) {

        Logger.info("create dao driver:"+driverName+",jdbc:"+url+",userName:"+userName);
        String validSql="";
        DBType dbType=DBType.parseFromURL(url);
        if("oracle".equals(type)){
            validSql="select 1 from dual";
            dbType=DBType.ORACLE;
        }else if("db2".equals(type)){
            validSql="select 1 from sysibm.sysdummy1";
            dbType=DBType.DB2;
        }else if("mysql".equals(type)){
            validSql="select 1 ";
            dbType=DBType.MYSQL;
        }else if("sqlserver".equals(type)){
            dbType=DBType.SQLSVR;
            validSql="select 1 ";
        }else if("pg".equals(type)){
            validSql="select 1 ";
            dbType=DBType.PG;
        }else if("dm".equals(type)){
            validSql="select 1 from dual";
            dbType=DBType.DM;
        }
        if(dbType==null){
            dbType=DBType.parseFromURL(url);
        }
        if(dbType==null){
            dbType=DBType.parseFromDriver(url);
        }
        if(dbType==null){
            Logger.info("dbType is null,url:"+url);
            return null;
        }
        Logger.info("dbType:"+dbType.getDAOType());
        DruidDataSource dataSource = new DruidDataSource();
        dataSource.setDriverClassName(driverName);
        dataSource.setUrl(url);
        dataSource.setUsername(userName);
        dataSource.setPassword(passwd);
        dataSource.setMaxActive(3);
        dataSource.setPhyMaxUseCount(2L);
        dataSource.setPhyMaxUseCount(2L);

        if(!StringUtil.isBlank(validSql)){
            dataSource.setValidationQuery(validSql);
        }
        dataSource.setMaxWait(3000L);
        dataSource.setConnectionErrorRetryAttempts(2);
        dataSource.setBreakAfterAcquireFailure(true);
        if(dbType==DBType.ORACLE) {
            // mysql 关闭，Oracle 建议开启
            dataSource.setPoolPreparedStatements(true);
        }
        if(dbType==DBType.DM) {
            // mysql 关闭，Oracle 建议开启
            dataSource.setPoolPreparedStatements(true);
        }
        return dataSource;
    }

}
